/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package com.snda.dw.lidejia.partitionfilegenerator;

import java.io.IOException;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.ql.io.HiveKey;
import org.apache.hadoop.hive.serde2.binarysortable.OutputByteBuffer;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.LineRecordReader;
import org.apache.hadoop.mapred.RecordReader;

/**
 *
 * @author lidejia
 */
public class HiveKeyRecordReader implements RecordReader<HiveKey, NullWritable> {

    private LineRecordReader reader;
    private List<PartitionerMeta> metas;
    private OutputByteBuffer buffer = new OutputByteBuffer();

    public HiveKeyRecordReader(Configuration job, FileSplit split) throws IOException {
        reader = new LineRecordReader(job, split);
        //
        metas = PartitionerMeta.parseMetas(job);
    }

    public boolean next(HiveKey k, NullWritable v) throws IOException {
        LongWritable offset = new LongWritable();
        Text line = new Text();
        boolean next = reader.next(offset, line);
        //
        if (next) {
            String[] tokens = line.toString().split("\t");
            buffer.reset();
            for (PartitionerMeta meta : metas) {
                meta.parse(tokens, buffer);
            }
            k.set(buffer.getData(), 0, buffer.getLength());
        }
        //
        return next;
    }

    public HiveKey createKey() {
        return new HiveKey();
    }

    public NullWritable createValue() {
        return NullWritable.get();
    }

    public long getPos() throws IOException {
        return reader.getPos();
    }

    public void close() throws IOException {
        reader.close();
    }

    public float getProgress() throws IOException {
        return reader.getProgress();
    }
}
